package com.datareport.service;

import com.datareport.common.exception.BusinessException;
import com.datareport.domain.entity.ExtractConfig;
import com.datareport.repository.ExtractConfigRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.List;

/**
 * 数据抽取服务
 */
@Service
public class ExtractService {
    private static final Logger log = LoggerFactory.getLogger(ExtractService.class);
    @Autowired
    private ExtractConfigRepository configRepository;
    
    /**
     * 创建抽取配置
     * @param config 抽取配置
     * @return 配置ID
     */
    @Transactional
    public String createConfig(ExtractConfig config) {
        // 1. 验证配置有效性
        validateConfig(config);
        
        // 2. 测试数据源连接
        testConnection(config);
        
        // 3. 保存配置
        config.setStatus(1);
        config.setCreateTime(LocalDateTime.now());
        config.setUpdateTime(LocalDateTime.now());
        configRepository.insert(config);
        
        log.info("创建抽取配置成功, 配置ID: {}", config.getConfigId());
        
        // 4. 触发首次全量抽取
        triggerFullExtract(config);
        
        return config.getConfigId();
    }
    
    /**
     * 更新抽取配置
     * @param config 抽取配置
     */
    @Transactional
    public void updateConfig(ExtractConfig config) {
        // 1. 检查配置是否存在
        ExtractConfig existConfig = configRepository.selectById(config.getConfigId());
        if (existConfig == null) {
            throw new BusinessException("抽取配置不存在");
        }
        
        // 2. 验证配置有效性
        validateConfig(config);
        
        // 3. 测试数据源连接
        testConnection(config);
        
        // 4. 更新配置
        config.setUpdateTime(LocalDateTime.now());
        configRepository.updateById(config);
    }
    
    /**
     * 删除抽取配置
     * @param configId 配置ID
     */
    @Transactional
    public void deleteConfig(String configId) {
        ExtractConfig config = configRepository.selectById(configId);
        if (config == null) {
            throw new BusinessException("抽取配置不存在");
        }
        
        configRepository.deleteById(configId);
    }
    
    /**
     * 获取抽取配置详情
     * @param configId 配置ID
     * @return 抽取配置
     */
    public ExtractConfig getConfig(String configId) {
        ExtractConfig config = configRepository.selectById(configId);
        if (config == null) {
            throw new BusinessException("抽取配置不存在");
        }
        
        return config;
    }
    
    /**
     * 获取所有抽取配置
     * @return 抽取配置列表
     */
    public List<ExtractConfig> listConfigs() {
        return configRepository.selectList(null);
    }
    
    /**
     * 触发增量抽取
     * @param configId 配置ID
     */
    public void triggerIncrementalExtract(String configId) {
        ExtractConfig config = configRepository.selectById(configId);
        if (config == null || config.getStatus() == 0) {
            throw new BusinessException("抽取配置不存在或已禁用");
        }
        
        // 构建Flink任务
        log.info("触发增量抽取任务, 配置ID: {}", configId);
        
        // 提交到Flink集群
        submitToFlinkCluster(config, false);
    }
    
    /**
     * 验证配置有效性
     * @param config 抽取配置
     */
    private void validateConfig(ExtractConfig config) {
        if (config.getSourceName() == null || config.getSourceName().isEmpty()) {
            throw new BusinessException("数据源名称不能为空");
        }
        
        if (config.getSourceType() == null || config.getSourceType().isEmpty()) {
            throw new BusinessException("数据源类型不能为空");
        }
        
        if (config.getSourceConfig() == null || config.getSourceConfig().isEmpty()) {
            throw new BusinessException("数据源配置不能为空");
        }
        
        if (config.getExtractTable() == null || config.getExtractTable().isEmpty()) {
            throw new BusinessException("抽取表名不能为空");
        }
        
        if (config.getExtractFields() == null || config.getExtractFields().isEmpty()) {
            throw new BusinessException("抽取字段不能为空");
        }
    }
    
    /**
     * 测试数据源连接
     * @param config 抽取配置
     */
    private void testConnection(ExtractConfig config) {
        try {
            // 实际项目中应该根据配置连接数据源进行测试
            log.info("测试数据源连接: {}", config.getSourceName());
        } catch (Exception e) {
            log.error("数据源连接测试失败", e);
            throw new BusinessException("数据源连接测试失败: " + e.getMessage());
        }
    }
    
    /**
     * 触发全量抽取
     * @param config 抽取配置
     */
    private void triggerFullExtract(ExtractConfig config) {
        log.info("触发全量抽取任务, 配置ID: {}", config.getConfigId());
        
        // 提交到Flink集群
        submitToFlinkCluster(config, true);
    }
    
    /**
     * 提交到Flink集群
     * @param config 抽取配置
     * @param isFullExtract 是否全量抽取
     */
    private void submitToFlinkCluster(ExtractConfig config, boolean isFullExtract) {
        // 实际项目中应该构建Flink任务并提交到集群
        log.info("提交{}抽取任务到Flink集群, 配置ID: {}", isFullExtract ? "全量" : "增量", config.getConfigId());
    }
}